package com.bw.yk03;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class FlinkDWDAsyncIO extends RichAsyncFunction<OrderWide,OrderWide> {
    private Connection conn = null;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;
    PreparedStatement ps3 = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        ps1 = conn.prepareStatement("select line from dim_user_info where id = ?");
        ps2 = conn.prepareStatement("select line from dim_sku_info where id = ?");
        ps3 = conn.prepareStatement("select line from dim_base_province where id = ?");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        ps1.close();
        ps2.close();
        ps3.close();
        conn.close();
    }
    @Override
    public void asyncInvoke(OrderWide orderWide, ResultFuture<OrderWide> resultFuture) throws Exception {
        Long user_id = orderWide.getOrderInfo().getUser_id();
        Long sku_id = orderWide.getOrderDetail().getSku_id();
        Long province_id = orderWide.getOrderInfo().getProvince_id();
        //查询dim_user_info
        ps1.setString(1,user_id+"");
        ResultSet rs1 = ps1.executeQuery();
        if(rs1.next()){
            String line = rs1.getString("line");
            UserInfo userInfo = JSON.parseObject(line, UserInfo.class);
            orderWide.setUserInfo(userInfo);
        }
        //查询dim_sku_info
        ps2.setString(1,sku_id+"");
        ResultSet rs2 = ps2.executeQuery();
        if(rs2.next()){
            String line = rs2.getString("line");
            SkuInfo skuInfo = JSON.parseObject(line, SkuInfo.class);
            orderWide.setSkuInfo(skuInfo);
        }
        //查询dim_base_province
        ps3.setString(1,province_id+"");
        ResultSet rs3 = ps3.executeQuery();
        if(rs3.next()){
            String line = rs3.getString("line");
            BaseProvince province = JSON.parseObject(line, BaseProvince.class);
            orderWide.setBaseProvince(province);
        }

        //返回宽表
        resultFuture.complete(Collections.singleton(orderWide));
    }
}
